package com.shujia.spark.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}

object Demo11CarBon {
  def main(args: Array[String]): Unit = {

    val spark: SparkSession = SparkSession
      .builder()
      .master("local")
      .appName("udf")
      .config("spark.sql.shuffle.partitions", 1)
      .getOrCreate()

    import spark.implicits._
    import org.apache.spark.sql.functions._


    //1、读取数据
    val lowCarBon: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "\t")
      .schema("user_id STRING , data_dt STRING ,low_carbon INT")
      .load("data/ant_user_low_carbon.txt")

    val plantCarBon: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "\t")
      .schema("plant_id STRING , plant_name STRING ,plant_carbon INT")
      .load("data/ant_plant_carbon.txt")


    lowCarBon.show()
    plantCarBon.show()


    /**
      * 蚂蚁森林植物申领统计
      * 问题：假设2017年1月1日开始记录低碳数据（user_low_carbon）,假设2017年10月1日之前满足申领条件的用户都申领了一颗p004-胡杨，
      * 剩余的能量全部用来领取“p002-沙柳” 。
      * 统计在10月1日累计申领“p002-沙柳” 排名前10的用户信息；以及他比后一名多领了几颗沙柳。
      * 得到的统计结果如下表样式：
      *
      *
      * crossJoin: 笛卡尔积
      */

    //取出胡杨树所需要的克数
    val huyang: DataFrame = plantCarBon.where($"plant_id" === "p004")
    //取出沙柳树所需要的克数
    val shaliu: DataFrame = plantCarBon.where($"plant_id" === "p002")

    lowCarBon
      //取出满足条件的数据
      .where($"data_dt" < "2017/10/1" and $"data_dt" >= "2017/1/1")
      //计算每个用户一年总的减少碳排放量
      .groupBy($"user_id")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //笛卡尔积
      .crossJoin(huyang)
      //统计用户剩余的克数
      .select($"user_id", $"sum_low_carbon" - $"plant_carbon" as "sy_low_carbon")
      //笛卡尔积
      .crossJoin(shaliu)
      //计算用户可以领取多少可沙柳
      .withColumn("plant_count", floor($"sy_low_carbon" / $"plant_carbon"))
      //获取后一名领取的数量
      .withColumn("less_count", lead($"plant_count", 1, 0) over Window.orderBy($"plant_count".desc))
      .withColumn("less_num", $"plant_count" - $"less_count")
      .limit(10)
      .select($"user_id", $"plant_count", $"less_num")
      .show()


    /**
      *
      * 问题：查询user_low_carbon表中每日流水记录，条件为：
      * 用户在2017年，连续三天（或以上）的天数里，
      * 每天减少碳排放（low_carbon）都超过100g的用户低碳流水。
      *
      */

    lowCarBon
      .where(year(regexp_replace($"data_dt", "/", "-")) === "2017")
      //计算米格用户每一天总分克数
      .groupBy($"user_id", $"data_dt")
      .agg(sum($"low_carbon") as "sum_low_carbon")
      //超过100g的用户
      .where($"sum_low_carbon" > 100)
      //增加排名字段
      .withColumn("r", row_number() over Window.partitionBy($"user_id").orderBy($"data_dt"))
      //使用日期减去row_number这一个列
      .withColumn("clazz", expr("date_sub(regexp_replace(data_dt,'/','-'),r)"))
      //统计连续的天数
      .withColumn("clazz_count", count($"user_id") over Window.partitionBy($"user_id", $"clazz"))
      //取出连续3天以上的
      .where($"clazz_count" >= 3)
      //关联原始的流水数据
      .join(lowCarBon, Seq("user_id", "data_dt"))
      .select($"user_id", $"data_dt", $"low_carbon")
      .show()


  }

}
